package com.mj.consumer;

import com.mj.utils.Util;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.store.ReadOffsetType;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * 拉模式，消费者主动拉取
 */
public class PullConsumer {
    public static void main(String[] args) throws MQClientException {
        // 1.和生产者一样，创建一个DefaultMQPushConsumer实例
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(Util.ConsumerGroup);
        // 2.设置NameServer的地址，消费者通过这个地址与NameServer进行通信，来获取Broker的地址信息
        consumer.setNamesrvAddr(Util.NamesrvAddr);
        Set<String> topics = new HashSet<>();
        topics.add(Util.Topic);
        consumer.setRegisterTopics(topics);
        // 3.启动消费者实例
        consumer.start();
        System.out.println("消费者已启动");
        // 4.获取每个topic里的信息
        while (true) {
            consumer.getRegisterTopics().forEach(topic -> {
                try {
                    Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(topic);
                    messageQueues.forEach(messageQueue -> {
                        try {
                            //通过本地内存作为缓存进行读取偏移量
                            long offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY);
                            if (offset < 0) {
                                offset = consumer.getOffsetStore().readOffset(messageQueue, ReadOffsetType.READ_FROM_STORE);
                            }
                            //此时可能没有偏移量，返回值就会是-1
                            if (offset < 0) {
                                //偏移量设置为最大偏移量
                                offset = consumer.maxOffset(messageQueue);
                            }
                            //但是也可能没有最大偏移量，就直接定义为0
                            if (offset < 0) {
                                offset = 0;
                            }
                            //一次拉取32条
                            PullResult pullResult = consumer.pull(messageQueue, "*", offset, 32);
                            System.out.println("消息拉取成功:" + pullResult.getPullStatus());
                            //判断拉取结果是否成功
                            switch (pullResult.getPullStatus()) {
                                case FOUND:
                                    pullResult.getMsgFoundList().forEach(msg -> {
                                        System.out.printf("消息：" + msg);
                                    });
                                    //自己维护偏移量
                                    consumer.updateConsumeOffset(messageQueue, pullResult.getNextBeginOffset());
                            }
                        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
                            e.printStackTrace();
                        }
                    });
                } catch (MQClientException e) {
                    e.printStackTrace();
                }
            });
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
